package com.alibaba.alink.operator.common.feature.binning;

import com.alibaba.alink.common.utils.Functional;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.util.Preconditions;

import java.io.Serializable;

/**
 * Some types defined for Binning.
 */
public class BinTypes {
    /**
     * DivideType for FeatureBorder.
     */
    public enum BinDivideType {
        /**
         * The FeatureBorder is generated by QuantileDiscretizerTrainBatchOp.
         */
        QUANTILE,

        /**
         * The FeatureBorder is generated by EqualWidthDiscretizerTrainBatchOp.
         */
        BUCKET,

        /**
         * The FeatureBorder is generated by OneHotTrainBatchOp.
         */
        DISCRETE,

        /**
         * The FeatureBorder is use defined.
         */
        USER_DEFINE
    }

    /**
     * The infType of SingleBorder.
     */
    public enum InfType {
        /**
         * The SingleBorder is -inf.
         */
        NEGATIVE_INF,

        /**
         * The SingleBorder is +inf.
         */
        POSITIVE_INF,

        /**
         * The SingleBorder is not infinity.
         */
        NOT_INF
    }

    public enum ColType {
        /**
         * Long type, for Int/Short/Byte/Long data.
         */
        LONG(Object::toString, true),

        /**
         * String type, for String/Boolean/Char.
         */
        STRING(Object::toString, false),

        /**
         * Double type, for Double/Float.
         */
        DOUBLE(Object::toString, true);

        final Functional.SerializableFunction<Object, String> objToStrFunc;
        final public boolean isNumeric;

        ColType(Functional.SerializableFunction<Object, String> objToStrFunc,
                boolean isNumeric) {
            this.objToStrFunc = objToStrFunc;
            this.isNumeric = isNumeric;
        }

        public static ColType valueOf(TypeInformation type) {
            if (type.equals(Types.DOUBLE) || type.equals(Types.FLOAT)) {
                return DOUBLE;
            } else if (type.equals(Types.LONG) || type.equals(Types.INT) || type.equals(Types.SHORT) || type.equals(Types.BYTE)) {
                return LONG;
            } else if (type.equals(Types.STRING) || type.equals(Types.BOOLEAN) || type.equals(Types.CHAR)) {
                return STRING;
            } else {
                throw new IllegalArgumentException("Unsupported type: " + type);
            }
        }
    }

    /**
     * Symbol for SingleBorder.
     */
    public enum Symbol {
        /**
         * Equal to the obj.
         */
        EQUAL,

        /**
         * Less than the obj.
         */
        LESS,

        /**
         * Less than or equal to the obj.
         */
        LESS_EQUAL,

        /**
         * Greater than the obj.
         */
        GREATER,

        /**
         * Greater than or equal to the obj.
         */
        GREATER_EQUAL
    }

    /**
     * The SingleBorder for FeatureBorder.
     * For discrete FeatureBorder, the SingleBorder must be (Symbol.EQUAL, record, InfType.
     */
    public static class SingleBorder implements Serializable {
        Symbol symbol;
        Object record;
        InfType infType;

        public SingleBorder(Symbol symbol, InfType infType) {
            this(symbol, null, infType);
        }

        public SingleBorder(Symbol symbol, Object record) {
            this(symbol, record, InfType.NOT_INF);
        }

        private SingleBorder(Symbol symbol, Object record, InfType infType) {
            Preconditions.checkNotNull(symbol, "Input Symbol is NULL!");
            Preconditions.checkNotNull(infType, "Input InfType is NULL!");

            this.symbol = symbol;
            this.record = record;
            this.infType = infType;
        }

        @Override
        public boolean equals(Object obj) {
            if (!(obj instanceof SingleBorder)) {
                return false;
            }
            SingleBorder singleBorder = (SingleBorder)obj;
            return this.symbol.equals(singleBorder.symbol) &&
                this.infType.equals(singleBorder.infType) &&
                BinningUtil.objectComparator.compare(this.record, singleBorder.record) == 0;
        }

        @Override
        public int hashCode(){
            return 0;
        }

        public Object getRecord() {
            return record;
        }

        public boolean isOpen() {
            return symbol.equals(Symbol.GREATER) || symbol.equals(Symbol.LESS);
        }
    }
}
